package com.tang;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * @author tmh
 * @date 2024/3/12 16:21
 * @description
 */
public class Subscribe03_Multi_Consumer {
    public static void main(String[] args) throws Exception{
        //1.创建消费者对象
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("Subscribe03_Multi_Consumer");

        //2.为消费者对象创建nameServer的地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //3.订阅主题
        consumer.subscribe("Filter-Test-Topic","TG1 || TG3");
        //4.注册监听消息，并且打印消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext context) {
                for (MessageExt msg : list) {
                    Date date = new Date();
                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    String acceptMsgTime = simpleDateFormat.format(date);
                    String s = new String(msg.getBody()) + " 接受到消息的时间为：" + acceptMsgTime;
                    System.out.println(s);
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        //4.把消费者直接启动起来
        consumer.start();
        System.out.println("consumer启动成功");
    }
}
